Reduce memory usage of HiveSplitSource #9232

haozhun · 2017-10-28T00:19:58Z

The first 3 commits introduces a new operation to AsyncQueue and some other preparatory work. Can you please review, @dain?

@electrum, can you please review the last 3 commits?

The 4th commit ("Reduce memory usage of HiveSplitSource") utilize the new operation to reduce memory usage of HiveSplitSource.
The 5th commit ("Reduce loader concurrency in BackgroundHiveSplitLoader") fixes a minor inefficiency.
The 6th commit is a cleanup in Hive connector that @electrum asked me to do

dain

First three look good... some minor comments

dain · 2017-10-30T20:02:14Z

presto-hive/src/main/java/com/facebook/presto/hive/util/AsyncQueue.java

-        else {
-            return Futures.transform(notEmptySignal, x -> getBatch(maxSize), executor);
+        ListenableFuture<List<T>> borrowedListFuture;
+        AtomicBoolean borrowerIncremented = new AtomicBoolean();


I don't think you need this. You only increment when the list is not empty so you can just check that in the transform below

dain · 2017-10-30T20:03:19Z

presto-hive/src/main/java/com/facebook/presto/hive/util/AsyncQueue.java

+                    try {
+                        BorrowResult<T, O> borrowResult;
+                        borrowResult = function.apply(elements);
+                        if (!borrowerIncremented.get()) {


if (elements.isEmpty)

dain · 2017-10-30T20:06:25Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+        assertEquals(queue.getBatchAsync(100).get(), ImmutableList.of("1", "2", "3"));
+
+        queue.finish();
+        assertTrue(queue.isFinished());


verify get after finish and maybe?

This is tested in testOfferAfterFinish

dain · 2017-10-30T20:08:10Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+        assertTrue(queue.offer("2").isDone());
+        assertTrue(queue.offer("3").isDone());
+
+        assertFalse(queue.offer("4").isDone());


if the queue is size 4, why doesn't this finish?

It's full once you insert the 4th one. The returned future tells you whether it's ready to accept additional inputs.

dain · 2017-10-30T20:10:10Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+        queue.offer("4");
+
+        queue.finish();
+        queue.offer("5");


check return value

dain · 2017-10-30T20:18:30Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+        queue.offer(4);
+        queue.offer(5);
+
+        Runnable runnable = () -> {


add a comment... remove up to three elements and reinsert them 700 times

dain · 2017-10-30T20:19:18Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+
+        AtomicBoolean done = new AtomicBoolean();
+        executor.submit(() -> {
+            if (!done.get()) {


dain · 2017-10-30T20:19:50Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+        AtomicBoolean done = new AtomicBoolean();
+        executor.submit(() -> {
+            if (!done.get()) {
+                assertFalse(queue.isFinished());


I'm pretty sure this is a race

assertFalse(queue.isFinished() || !done.get()

dain · 2017-10-30T20:21:57Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+
+        assertFalse(queue.isFinished());
+        ArrayList<Integer> list = new ArrayList<>(queue.getBatchAsync(100).get());
+        // 1 and 2 were removed by borrow call; 8 and 9 were never inserted because insertion happened after finish.


I'd add a separate test for " 8 and 9 were never inserted because insertion happened after finish"

This is tested in testOfferAfterFinish

haozhun

Addressed comments from @dain

haozhun · 2017-11-05T23:45:21Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+        assertTrue(queue.offer("2").isDone());
+        assertTrue(queue.offer("3").isDone());
+
+        assertFalse(queue.offer("4").isDone());


It's full once you insert the 4th one. The returned future tells you whether it's ready to accept additional inputs.

haozhun · 2017-11-06T00:22:00Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+
+        assertFalse(queue.isFinished());
+        ArrayList<Integer> list = new ArrayList<>(queue.getBatchAsync(100).get());
+        // 1 and 2 were removed by borrow call; 8 and 9 were never inserted because insertion happened after finish.


This is tested in testOfferAfterFinish

haozhun · 2017-11-06T00:33:41Z

presto-hive/src/test/java/com/facebook/presto/hive/util/TestAsyncQueue.java

+        assertEquals(queue.getBatchAsync(100).get(), ImmutableList.of("1", "2", "3"));
+
+        queue.finish();
+        assertTrue(queue.isFinished());


This is tested in testOfferAfterFinish

haozhun · 2017-11-06T00:37:09Z

@electrum Please review the last 3 commits. They should be pretty straightforward.

electrum · 2017-11-07T21:07:43Z

"Remove connectorId" looks good

electrum · 2017-11-07T21:08:41Z

"Reduce loader concurrency" looks good

electrum

I think InternalHiveSplit shouldn't have end

electrum · 2017-11-07T21:16:02Z

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveSplitSource.java

@@ -217,8 +222,9 @@ public void testOutstandingSplitSize()
                100,
                new Properties(),
                ImmutableList.of(new HivePartitionKey("pk_col", "pk_value")),
-                ImmutableList.of(HostAddress.fromString("localhost")),
+                ImmutableList.of(new InternalHiveSplit.InternalHiveBlock(0, 100, ImmutableList.of(HostAddress.fromString("localhost")))),


Qualifying InternalHiveBlock isn't needed

electrum · 2017-11-07T21:18:40Z

presto-hive/src/main/java/com/facebook/presto/hive/InternalHiveSplit.java

    public int getEstimatedSizeInBytes()
    {
+        // TODO! revisit


Add more detail or remove this comment. It's not useful to readers in the current form

electrum · 2017-11-07T21:21:09Z

presto-hive/src/main/java/com/facebook/presto/hive/InternalHiveSplit.java

+
+        public InternalHiveBlock(long start, long end, List<HostAddress> addresses)
+        {
+            checkArgument(start <= end, "block must not have negative length");


I'd say "block end cannot be before block start"

electrum · 2017-11-07T21:22:07Z

presto-hive/src/main/java/com/facebook/presto/hive/InternalHiveSplit.java

+            checkArgument(start <= end, "block must not have negative length");
+            this.start = start;
+            this.end = end;
+            this.addresses = addresses;


Defensive copy

electrum · 2017-11-07T21:22:27Z

presto-hive/src/main/java/com/facebook/presto/hive/InternalHiveSplit.java

+            int result = INSTANCE_SIZE;
+            result += SizeOf.sizeOfObjectArray(addresses.size());
+            for (HostAddress address : addresses) {
+                result += HOST_ADDRESS_INSTANCE_SIZE + address.getHostText().length() * Character.BYTES;


This is easier to read without paren

electrum · 2017-11-07T21:30:44Z

presto-hive/src/main/java/com/facebook/presto/hive/BackgroundHiveSplitLoader.java

+                continue;
+            }
+            if (blockStart == blockEnd && !(blockStart == start && blockEnd == start + length)) {
+                // skip zero-width block, except in the special circumstance: slice is empty, and the block covers the empty slice interval.


Why do we need a special case for this?

This seems to be a special case to generate a split for an empty file. Why not just check that up front?

electrum · 2017-11-07T21:31:50Z

presto-hive/src/main/java/com/facebook/presto/hive/BackgroundHiveSplitLoader.java

-                        int chunks = toIntExact((long) Math.ceil((blockLocation.getLength() - chunkOffset) * 1.0 / maxBytes));
-                        targetChunkSize = (long) Math.ceil((blockLocation.getLength() - chunkOffset) * 1.0 / chunks);
-                    }
+        // TODO: remainingInitialSplits, maxInitialSplitSize, maxSplitSize


What does this mean?

electrum · 2017-11-07T21:32:35Z

presto-hive/src/main/java/com/facebook/presto/hive/BackgroundHiveSplitLoader.java

+    {
+        checkArgument(!blocks.isEmpty());
+        checkArgument(start == blocks.get(0).getStart());
+        checkArgument(start + length == blocks.get(blocks.size() - 1).getEnd());


Why doesn't this fail for the first block?

I don't understand the question

electrum · 2017-11-07T21:34:36Z

presto-hive/src/main/java/com/facebook/presto/hive/BackgroundHiveSplitLoader.java

+        catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+        for (String host : hosts) {


return Arrays.stream(hosts) .map(HostAddress::fromString) .collect(toImmutableList());

electrum · 2017-11-07T21:36:04Z

presto-hive/src/main/java/com/facebook/presto/hive/HiveSplitSource.java

+            }
+            estimatedSplitSizeInBytes.addAndGet(-removedEstimatedSizeInBytes);
+
+            ImmutableList<InternalHiveSplit> splitsToInsert = splitsToInsertBuilder.build();


Declare as List

Previously, HiveSplitSource exposes oustanding split count by querying the size of the underlying AsyncQueue. The AsyncQueue.size method need to be removed as its definition will become ambiguous with addition of new method in the next commit.

Use each InternalHiveSplit to represent a file, instead of part of a file. Each InternalHiveSplit can now correspond to multiple HiveSplits. Responsibility to produce splits of appropriate size has been moved from BackgroundHiveSplitLoader to HiveSplitSource.

Previously, BackgroundHiveSplitLoader uses maxPartitionBatchSize as the number of maximum concurrent threads per split source. The default value for the configuration parameter is 100. This is wasteful and unnecessary. Before the previous rewrite of BackgroundHiveSplitLoader, the number of threads is not directly bounded, but is finite due to other constraints. At the time, the maximum possible number of threads possible were chosen.

vgankidi · 2018-10-26T00:15:55Z

@haozhun We're seeing some performance regression with the reduced loader concurrency in BackgroundHiveSplitLoader: https://groups.google.com/forum/#!topic/presto-users/AlEr1jq-R0E
Although this config is tunable, what do you think about bumping the default value a bit higher? 4 split loader threads per HiveSplitSource seem low for s3 listings.

haozhun assigned electrum and dain Oct 28, 2017

facebook-github-bot added the CLA Signed label Oct 28, 2017

haozhun force-pushed the hive-split branch from 97fcf58 to 5c00c42 Compare October 28, 2017 00:20

dain reviewed Oct 30, 2017

View reviewed changes

dain removed their assignment Oct 30, 2017

haozhun commented Nov 6, 2017

View reviewed changes

haozhun force-pushed the hive-split branch from 5c00c42 to 198bcae Compare November 6, 2017 00:36

electrum reviewed Nov 7, 2017

View reviewed changes

haozhun added 3 commits November 7, 2017 14:44

Replace CompletableFuture with ListenableFuture in AsyncQueue

ece4ece

Add borrow method to AsyncQueue

7b4ca22

haozhun force-pushed the hive-split branch 2 times, most recently from e9bc73e to da6fee1 Compare November 7, 2017 23:05

haozhun added 3 commits November 7, 2017 15:10

Remove connectorId/clientId from most Hive connector entities

504fae0

haozhun force-pushed the hive-split branch from da6fee1 to 504fae0 Compare November 7, 2017 23:10

electrum approved these changes Nov 7, 2017

View reviewed changes

haozhun merged commit 504fae0 into prestodb:master Nov 7, 2017

This was referenced Nov 8, 2017

Set MALLOC_ARENA_MAX=2 for product tests docker Java containers #9310

Closed

Fix handling of empty files in Hive connector #9312

Closed

tpch.q17 product tests fails due exceeding travis memory limit #9295

Closed

wenleix mentioned this pull request Apr 6, 2019

Bound total memory used by HiveSplitSource #9119

Merged

Reduce memory usage of HiveSplitSource #9232

Reduce memory usage of HiveSplitSource #9232

Conversation

haozhun commented Oct 28, 2017 • edited by findepi Loading

dain left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

haozhun left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

haozhun commented Nov 6, 2017

electrum commented Nov 7, 2017

electrum commented Nov 7, 2017

electrum left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vgankidi commented Oct 26, 2018

haozhun commented Oct 28, 2017 •

edited by findepi

Loading